package com.im.config;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import com.im.pojo.*;
import com.im.pojoClass.ImLatelyList;
import com.im.pojoClass.ImSendList;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.List;

//@Async
@Component
@Slf4j
public class ClientSample implements CommandLineRunner {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private StringUtil stringUtil;

//    private static HashMap<String, Object> message = new HashMap<>();

    @Override
    public void run(String... args) throws Exception {
        main();
    }

    @Async
    public void main() {
        // 创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("110.42.132.69", 11111), "example", "", "");
        int batchSize = 1000;
        while (true) {
            try {
                connector.connect();
                connector.subscribe(".*\\..*");
                connector.rollback();
            } catch (Exception e) {
                log.info("链接失败，10秒后尝试重新链接");
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException interruptedException) {
                    interruptedException.printStackTrace();
                }
                continue;
            }
            log.info("Canal连接成功");
            break;
        }
        try {

            while (true) {
                // 获取指定数量的数据
                Message message = connector.getWithoutAck(batchSize);
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    printEntry(message.getEntries(), connector, batchId);
                }

                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            connector.disconnect();
        }
    }

    //    @Bean
    @Async
    public void printEntry(List<Entry> entrys, CanalConnector connector, long batchId) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }

            RowChange rowChage = null;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
            }

            //entry.getHeader().getLogfileName();//binlog文件
            //entry.getHeader().getLogfileOffset();//binlog偏移量

            EventType eventType = rowChage.getEventType();
            String schemaName = entry.getHeader().getSchemaName();//数据库
            String tableName = entry.getHeader().getTableName();//数据表

            if (schemaName.equals("im") && eventType != EventType.QUERY) {
                JSONObject json = getJson(rowChage, eventType);
                HashOperations<String, String, String> stringHashOperations = redisTemplate.opsForHash();
                switch (tableName) {
                    case "im_user": {
                        ImUser imUser = json.toJavaObject(ImUser.class);
                        imUser.setUser_pass(null);
                        if (eventType == EventType.INSERT) {
                            stringHashOperations.put("im_user", imUser.getUser_id(), JSONObject.toJSONString(imUser));
                        }
                        break;
                    }
                    case "im_message": {
                        ImMessage imMessage = json.toJavaObject(ImMessage.class);
                        if (eventType == EventType.INSERT) {
                            stringHashOperations.put("im_message", imMessage.getMessage_id(), JSONObject.toJSONString(imMessage));
                        }
                        break;
                    }
                    case "im_lately": {
                        ImLately imLately = json.toJavaObject(ImLately.class);
                        if (eventType == EventType.INSERT || eventType == EventType.UPDATE) {
                            // 获取消息表
                            String im_message = stringHashOperations.get("im_message", imLately.getMessage_id());
                            ImMessage imMessage = JSONObject.parseObject(im_message, ImMessage.class);

                            // 获取用户表
                            String im_user = stringHashOperations.get("im_user", imLately.getReceive_id());
                            ImUser imUser = JSONObject.parseObject(im_user, ImUser.class);

                            ImLatelyList imLatelyList = new ImLatelyList(imLately, imUser, imMessage);

                            String send_id = "im_lately" + imLately.getSend_id();
                            String receive_id = imLately.getReceive_id();
                            String s = JSONObject.toJSONString(imLatelyList);
                            redisTemplate.opsForHash().put(send_id, receive_id, s);
                        }
                        break;
                    }
                    case "im_send": {
                        ImSend imSend = json.toJavaObject(ImSend.class);
                        if (eventType == EventType.INSERT) {
                            // 获取用户消息
                            String im_message = stringHashOperations.get("im_message", imSend.getMessage_id());
                            ImMessage imMessage = JSONObject.parseObject(im_message, ImMessage.class);

                            ImSendList imSendList = new ImSendList(imMessage, imSend);

                            String redis_id_key = stringUtil.str_compare(imSend.getSend_id(), imSend.getReceive_id());
                            redisTemplate.opsForZSet().add("im_send" + redis_id_key, JSONObject.toJSONString(imSendList), Double.parseDouble(imMessage.getMessage_date()));
                        }
                        break;
                    }
                    case "im_apply": {
                        ImApply imApply = json.toJavaObject(ImApply.class);
                        if (eventType == EventType.INSERT) {
                            // 获取用户表
                            String im_user = stringHashOperations.get("im_user", imApply.getSend_id());
                            stringHashOperations.put("im_apply" + imApply.getReceive_id(), imApply.getSend_id(), im_user);
                        }
                        if (eventType == EventType.UPDATE) {
                            stringHashOperations.delete("im_apply" + imApply.getReceive_id(), imApply.getSend_id());
                        }
                        break;
                    }
                    default: {
                        break;
                    }
                }
            }

        }
    }

    private static JSONObject getJson(RowChange rowChage, EventType eventType) {

        JSONObject jsonObject = new JSONObject();

        for (RowData r : rowChage.getRowDatasList()) {
            if (eventType == EventType.DELETE) {
                for (Column c : r.getBeforeColumnsList()) {
                    jsonObject.put(c.getName(), c.getValue());
                }
            } else {
                for (Column c : r.getAfterColumnsList()) {
                    jsonObject.put(c.getName(), c.getValue());
                }
            }
        }

        return jsonObject;
    }
}
